package com.qianbao.finance.sync.modules.elasticsearch.websocket.subscribe;


import com.qianbao.finance.sync.common.utils.SpringContextUtils;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.server.ChatRoom;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.server.ShareBean;
import com.qianbao.finance.sync.modules.elasticsearch.websocket.server.ShareConfig;
import org.apache.log4j.Logger;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;



/**
 * ZeroMQ 接收订阅消息
 * 
 * @author WangYuan
 * @date 2016年7月12日下午3:12:30
 */
public class Subscriber implements Runnable {

    private String spliter = "@";
    private Socket server;
    private Context context;
    private String dataBody = "";
    private String groupName = "";
    private boolean running = true;
    private String chatRoomName = "";

    private static String socketAddress = ((ShareConfig) SpringContextUtils.getBean("shareConfig")).getSubscriberAddress();

    private static final Logger logger = Logger.getLogger(Subscriber.class);

    @Override
    public void run() {
        ChatRoom chatRoom = null;

        try {
            context = ZMQ.context(1);
            logger.info("socket-server Subscriber上下文初始化成功");
            server = context.socket(ZMQ.SUB);
            server.subscribe("".getBytes());
            server.connect("tcp://localhost:7878");
            logger.info("成功订阅监听:" + socketAddress);
            while (running) {
                try {
                    byte[] data = server.recv(0);
                    if (logger.isDebugEnabled()) {
                        logger.debug("socket server got: " + new String(data));
                    }
                    if (null != data && data.length > 0) {
                        String dataStr = new String(data);
                        dataBody = dataStr.split(spliter)[2];
                        groupName = dataStr.split(spliter)[1];
                        chatRoomName = dataStr.split(spliter)[0];
                        chatRoom = ShareBean.getInstance().getChatRoom().get(chatRoomName);
                        if (null == chatRoom) {// 前台没有Socket Client链接 不推送消息
                            continue;
                        }
                        chatRoom.broadcastMessage(groupName, dataBody);
                        if (logger.isDebugEnabled()) {
                            logger.debug("socket server push: " + groupName + " " + chatRoomName
                                    + " " + dataBody);
                        }
                    }
                } catch (Exception e) {
                    logger.error("", e);
                }
            }
        } catch (Exception e) {
            logger.error("", e);
        }
    }

    public void doStop() {
        running = false;
        logger.info("停止成功");
    }

}
